package spark.work

import java.math.BigDecimal
import java.util.Date

import com.kongming.kmdm.model.KmDataColumnType
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, SQLContext, SparkSession}
import org.apache.spark.sql.functions.{col, lit, udf}
import org.apache.spark.sql.types.{DateType, DoubleType, LongType}
import org.apache.spark.sql._

import scala.util.Try

/**
  * Created by liuwei on 2018/1/10.
  */
object RowAppend {

  def main(args: Array[String]): Unit = {
    //    newLeft = newLeft.withColumn(name, colTypeTransform(newLeft.col(name)))
    val sparkConf = new SparkConf().setAppName("RowToColumnTest").setMaster("local[4]")
    val sc = new SparkContext(sparkConf)
    val sparkSession = SparkSession.builder.getOrCreate()
    val sqlContext = new SQLContext(sc)
    import sqlContext.implicits._
    var df = sqlContext.read.format("com.databricks.spark.csv").option("header", "true").option("inferSchema", true.toString).load(("C:\\Users\\lenovo\\Desktop\\huggies_hsfyf_user_info.csv")).repartition(6)
    val rdd = df.rdd.zipWithIndex().filter(_._2 < 5).map(_._1)
    df = sparkSession.createDataFrame(rdd, df.schema)

    df = df.withColumn("baby_birth", df.col("baby_birth").cast(LongType))
    df.count()
    println("----"+df.count())
    val newLeft = df.withColumn("baby_birth", colTypeTransform(df.col("baby_birth")))
    newLeft.show
//    val newNames = df.columns
//    println("===="+df.schema)
//    println("===="+newLeft.head().toSeq)
//    val outDf = getDf(df: DataFrame, newLeft: DataFrame, newNames.toArray, true)
//    outDf.show
//    println(outDf.schema)

  }


  private def getNewType(leftType: KmDataColumnType, rightType: KmDataColumnType): KmDataColumnType = {
    var _type = KmDataColumnType.TEXT
    if (leftType.equals(rightType)) _type = leftType
    _type
  }

  private def getDf(leftDf: DataFrame, rightDf: DataFrame, newNames: Array[String], typeConvert: Boolean): DataFrame = {
    var newLeft = leftDf
    var newRight = rightDf
    newNames.foreach(name => {
      val leftIdx = leftDf.schema.fieldNames.indexOf(name)
      val rightIdx = rightDf.schema.fieldNames.indexOf(name)
      if (-1 == leftIdx) {
        rightDf.schema.apply(name).dataType match {
          case DoubleType => newLeft = newLeft.withColumn(name, lit(null.asInstanceOf[Double]))
          case LongType => newLeft = newLeft.withColumn(name, lit(null.asInstanceOf[Long]))
          case _ => newLeft = newLeft.withColumn(name, lit(null: String))
        }
      } else if (-1 == rightIdx) {
        leftDf.schema.apply(name).dataType match {
          case DoubleType => newRight = newRight.withColumn(name, lit(null.asInstanceOf[Double]))
          case LongType => newRight = newRight.withColumn(name, lit(null.asInstanceOf[Long]))
          case _ => newRight = newRight.withColumn(name, lit(null: String))
        }
      } else if (typeConvert && (rightDf.schema.apply(name).dataType != leftDf.schema.apply(name).dataType)) {
        newLeft = newLeft.withColumn(name, colTypeTransform(newLeft.col(name)))
        newRight = newRight.withColumn(name, colTypeTransform(newRight.col(name)))
      }
      //      if (!leftDf.schema.fieldNames.contains(name)) newLeft = newLeft.withColumn(name, lit("": String))
      //      if (!rightDf.schema.fieldNames.contains(name)) newRight = newRight.withColumn(name, lit("": String))
      //      newLeft = newLeft.withColumn(name, colTypeTransform(newLeft.col(name)))
      //      newRight = newRight.withColumn(name, colTypeTransform(newRight.col(name)))
    })
    newLeft.select(newNames.head, newNames.tail: _*).union(newRight.select(newNames.head, newNames.tail: _*))
  }

  private def colTypeTransform =
    udf((value: Any) => {
      if (value.isInstanceOf[Double]) {
        Try(new BigDecimal(value.asInstanceOf[Double]).stripTrailingZeros.toPlainString).getOrElse(null)
      } else if (value.isInstanceOf[Long]) {
        import java.text.SimpleDateFormat
        val format=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
        Try(format.format(new java.util.Date(value.asInstanceOf[Long]))).getOrElse(null)
      } else {
        Try(value.toString).getOrElse(null)
      }
    })
}
